package com.yequan.mq.listener;

import com.alibaba.fastjson.JSON;
import com.yequan.exe.core.ExeExecutor;
import com.yequan.pojo.BaseCalcInfo;
import com.yequan.pojo.CalcData;
import com.yequan.pojo.ExeInfo;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * @author yq
 * @date 2020/09/29
 */
public class ExeMqMessageListener implements MessageListenerConcurrently {

    private ExeExecutor exeExecutor;

    private BaseCalcInfo baseCalcInfo;

    public ExeMqMessageListener(ExeExecutor exeExecutor, BaseCalcInfo baseCalcInfo) {
        this.exeExecutor = exeExecutor;
        this.baseCalcInfo = baseCalcInfo;
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        MessageExt messageExt = msgs.get(0);
        ExeInfo exeInfo = JSON.parseObject(messageExt.getBody(), ExeInfo.class);
        CalcData calcData = new CalcData();
        calcData.setBaseCalcInfo(baseCalcInfo);
        calcData.setExeInfo(exeInfo);
        exeExecutor.execute(calcData);
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

}
